#include "spdlog/spdlog.h"

#include "media_source.hpp"
#include "media_sink.hpp"
#include "transformat/media_bridge.hpp"

using namespace mms;

MediaSource::MediaSource(const std::string & media_type, std::weak_ptr<StreamSession> session, ThreadWorker * worker):
    media_type_(media_type), session_(session), worker_(worker)
{
}

MediaSource::~MediaSource()
{
}

bool MediaSource::add_media_sink(std::shared_ptr<MediaSink> media_sink)
{
    std::lock_guard<std::recursive_mutex> lck(sinks_mtx_);
    lazy_sinks_.insert(media_sink);
    media_sink->set_source(this->shared_from_this());
    return true;
}

bool MediaSource::remove_media_sink(std::shared_ptr<MediaSink> media_sink)
{
    std::lock_guard<std::recursive_mutex> lck(sinks_mtx_);
    for(auto it = lazy_sinks_.begin(); it != lazy_sinks_.end(); it++){
        if(*it == media_sink){
            lazy_sinks_.erase(it);
            break;
        }
    }
    spdlog::info("remove media sink syccess");
    return true;
}

std::shared_ptr<MediaBridge> mms::MediaSource::get_or_create_bridge(const std::string& id, StreamSession* session)
{
    return nullptr;
}

bool mms::MediaSource::remove_bridge(std::shared_ptr<MediaBridge> bridge)
{
    std::lock_guard<std::shared_mutex> lck(bridges_mtx_);
    for(auto it = bridges_.begin(); it != bridges_.end(); ++it){
        if(it->second == bridge){
            spdlog::info("find bridge : {} remove ", it->first);
            auto sink = it->second->get_media_sink();
            bridges_.erase(it);
            remove_media_sink(sink);
            last_sinks_or_bridges_leave_time_ = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
            break;
        }
    }
    return true;
}

bool mms::MediaSource::remove_bridge(const std::string& id)
{
    std::lock_guard<std::shared_mutex> lck(bridges_mtx_);
    auto it = bridges_.find(id);
    if (it == bridges_.end()) {
        return false;
    }
    auto sink = it->second->get_media_sink();
    bridges_.erase(id);
    remove_media_sink(sink);
    last_sinks_or_bridges_leave_time_ = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
    return true;
}

bool mms::MediaSource::is_no_sinks_for_time(uint32_t milli_sec)
{
    std::lock_guard<std::recursive_mutex> lck1(sinks_mtx_);
    std::shared_lock<std::shared_mutex> lck2(bridges_mtx_);
    if (lazy_sinks_.size() > 0 || bridges_.size() > 0) {
        return false;
    }

    int64_t now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now().time_since_epoch()).count();
    if (now_ms - last_sinks_or_bridges_leave_time_ < milli_sec) {
        return false;
    }
    return true;
}

std::shared_ptr<StreamSession> MediaSource::get_session()
{
   auto s = session_.lock();
   return s;
}

void MediaSource::set_source_info(const std::string &domain, const std::string &app_name, const std::string &stream_name)
{
    domain_ = domain;
    app_name_ = app_name;
    stream_name_ = stream_name;
}

void MediaSource::set_session(std::shared_ptr<StreamSession> s)
{
    session_ = std::weak_ptr<StreamSession>(s);
}

bool MediaSource::is_ready()
{
    return stream_ready_;
}

void MediaSource::close()
{
}
